
#include "event.h"
#include "EZLoop.h"
#include "common/EZLog.h"
#include "common/EZUtil.h"
#include "event2/thread.h"
#include <assert.h>


namespace EZTcp{


EZLoop::EZLoop()
	: create_evbase_myself_(true)
	, notified_(false)
	, pending_functor_count_(0)
	, runing_(false)
{
#ifdef _WIN32 
	WSADATA wsa;
	WSAStartup(MAKEWORD(2,2),&wsa);

    evbase_ = event_base_new();
#else
	evbase_ = event_base_new();
#endif
	_init();
}

EZLoop::EZLoop(struct event_base* base)
	: evbase_(base)
	, create_evbase_myself_(false)
	, notified_(false)
	, pending_functor_count_(0)
	, runing_(false)
{
#ifdef _WIN32
	WSADATA wsa;
	WSAStartup(MAKEWORD(2,2),&wsa);
#endif
	_init();

	// When we build an EventLoop instance from an existing event_base
	// object, we will never call EventLoop::Run() method.
	// So we need to watch the task queue here.
	auto rc = watcher_->async_wait();
	assert(rc);
	if (!rc) {
		EZLOG_FATAL << "PipeEventWatcher init failed.";
	}
	runing_.store(true);
}


EZLoop::~EZLoop()
{
#ifdef _WIN32
	WSACleanup();
#endif
	if (evbase_ != nullptr && create_evbase_myself_) {
		event_base_free(evbase_);
		evbase_ = nullptr;
	}
}

void EZLoop::after_fork()
{
    int rc = event_reinit(evbase_);
    assert(rc == 0);

    if (rc != 0) {
        EZLOG_FATAL << "event_reinit failed!";
        return;
    }

    // We create EventLoopThread and initialize it in father process,
    // but we use it in child process.
    // If we have only one child process, everything goes well.
    //
    // But if we have multi child processes, something goes wrong.
    // Because EventLoop::watcher_ is created and initialized in father process
    // all children processes inherited father's pipe.
    //
    // When we use the pipe to do a notification in one child process
    // the notification may be received by another child process randomly.
    //
    // So we need to reinitialize the watcher_
    _init_notify_pipe_watcher();
}

void EZLoop::run()
{
	EZLOG_DEBUG;

	tid_ = std::this_thread::get_id(); // The actual thread id

	int rc = watcher_->async_wait();
	assert(rc);
	if (!rc) {
		EZLOG_FATAL << "PipeEventWatcher init failed.";
	}

	runing_.store(true);
	rc = event_base_dispatch(evbase_);
	if (rc == 1) {
		EZLOG_ERROR << "event_base_dispatch error: no event registered";
	}
	else if (rc == -1) {
		int serrno = errno;
		EZLOG_ERROR << "event_base_dispatch error " << serrno << " "
			<< EZUtil::strerror(serrno);
	}

	// Make sure watcher_ does construct, initialize and destruct in the same thread.
	watcher_.reset();
	EZLOG_DEBUG << "EventLoop stopped, tid=" << std::this_thread::get_id();

}

void EZLoop::stop()
{
	EZLOG_DEBUG << "EventLoop::Stop";
	post(std::bind(&EZLoop::_stop_in_loop, this));
}

void EZLoop::dispatch(const void_0_arg_fun& handler)
{
	//EZLOG_DEBUG;
	if (is_running() && is_in_loop_thread()) {
		handler();
	}
	else {
		post(handler);
	}
}

void EZLoop::post(const void_0_arg_fun& handler)
{
	//EZLOG_DEBUG << "pending_functor_count_=" << pending_functor_count_ << " PendingQueueSize=" << _get_pending_queue_size() << " notified_=" << notified_.load();
	{
		std::lock_guard<std::mutex> lock(mutex_);
		pending_functors_.emplace_back(handler);
	}
	++pending_functor_count_;
	//EZLOG_DEBUG << "queued a new Functor. pending_functor_count_=" << pending_functor_count_ << " PendingQueueSize=" << _get_pending_queue_size() << " notified_=" << notified_.load();
	if (!notified_.load()) {
		//EZLOG_DEBUG << "call watcher_->Nofity() notified_.store(true)";

		// We must set notified_ to true before calling `watcher_->Nodify()`
		// otherwise there is a change that:
		//  1. We called watcher_- > Nodify() on thread1
		//  2. On thread2 we watched this event, so wakeup the CPU changed to run this EventLoop on thread2 and executed all the pending task
		//  3. Then the CPU changed to run on thread1 and set notified_ to true
		//  4. Then, some thread except thread2 call this QueueInLoop to push a task into the queue, and find notified_ is true, so there is no change to wakeup thread2 to execute this task
		notified_.store(true);

		// Sometimes one thread invoke EventLoop::QueueInLoop(...), but anther
		// thread is invoking EventLoop::Stop() to stop this loop. At this moment
		// this loop maybe is stopping and the watcher_ object maybe has been
		// released already.
		if (watcher_) {
			watcher_->notify();
		}
		else {
			EZLOG_DEBUG << "runing =" << runing_;
			assert(!is_running());
		}
	}
	// "No need to call watcher_->Nofity()";
}

void EZLoop::_init()
{
	tid_ = std::this_thread::get_id(); // The default thread id

	_init_notify_pipe_watcher();
}

void EZLoop::_init_notify_pipe_watcher()
{
	watcher_.reset(new PipeEventWatcher(this, std::bind(&EZLoop::_do_pending_functors, this)));
	int rc = watcher_->init();
	assert(rc);
	if (!rc) {
		EZLOG_FATAL << "PipeEventWatcher init failed.";
	}
}

void EZLoop::_stop_in_loop()
{
	EZLOG_DEBUG << "EventLoop is stopping now, tid=" << std::this_thread::get_id();

	auto f = [this]() {
		for (int i = 0;; i++) {
			EZLOG_DEBUG << "calling DoPendingFunctors index=" << i;
			_do_pending_functors();
			if (_is_pending_queue_empty()) {
				break;
			}
		}
	};

	EZLOG_DEBUG << "before event_base_loopexit, we invoke DoPendingFunctors";

	f();

	EZLOG_DEBUG << "start event_base_loopexit";
	runing_.store(false);
	event_base_loopexit(evbase_, nullptr);
	EZLOG_DEBUG << "after event_base_loopexit, we invoke DoPendingFunctors";

	f();

	EZLOG_DEBUG << "end of StopInLoop";
}

void EZLoop::_do_pending_functors()
{
	std::vector<void_0_arg_fun> functors;
	{
		std::lock_guard<std::mutex> lock(mutex_);
		notified_.store(false);
		pending_functors_.swap(functors);
	}
	for (size_t i = 0; i < functors.size(); ++i) {
		functors[i]();
		--pending_functor_count_;
	}
}


}
